package com.example.config;

import com.example.entity.Person;
import com.example.listener.BatchJobListener;
import com.example.listener.BatchReaderListener;
import com.example.listener.BatchWriterListener;
import com.example.processer.BatchItemValidateProcessor;
import com.example.validate.DataValidate;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.validator.Validator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * @description spring batch config
 * @author: ruanchengzhi
 * @create: 2021-07-29 19:50
 **/
@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Bean
    public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDataSource(dataSource);
        jobRepositoryFactoryBean.setTransactionManager(transactionManager);
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        return jobRepositoryFactoryBean.getObject();
    }

    @Bean
    public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
        return jobLauncher;
    }

    @Bean
    public Job importJob(JobBuilderFactory jobs, Step batchStep) {
        return jobs.get("batchJob")
                .incrementer(new RunIdIncrementer())
                .flow(batchStep)
                .end()
                .listener(batchJobListener())
                .build();
    }


    /**
     * Chunk的机制(即每次读取一条数据，再处理一条数据，累积到一定数量后再一次性交给writer进行写入操作。
     * retryLimit(3).retry(Exception.class) 重试机制 重试三次
     * skip(Exception.class).skipLimit(2)  跳过
     *
     * @param stepBuilderFactory
     * @return org.springframework.batch.core.Step
     * @throws
     * @author ruanchengzhi
     * @date 2021/7/30
     */
    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader, ItemWriter<Person> writer) {
        return stepBuilderFactory
                .get("batchStep")
                .<Person, Person>chunk(65000)
                .reader(reader).faultTolerant().retryLimit(3).retry(Exception.class).skip(Exception.class).skipLimit(2)
                .listener(new BatchReaderListener())
                .processor(validateProcessor())
                .writer(writer).faultTolerant().skip(Exception.class).skipLimit(2)
                .listener(new BatchWriterListener())
                .build();
    }



    /*@Bean
    public ItemReader<Person> reader() {
        return new BatchReader();
    }*/

    /*@Bean
    public ItemWriter<Person> writer() {
        return new BatchWriter();
    }*/

    /**
     * ItemReader定义：读取文件数据+entirty实体类映射
     *
     * @return
     */
    @Bean
    public ItemReader<Person> reader() {
        // 使用FlatFileItemReader去读cvs文件，一行即一条数据
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        // 设置文件处在路径
        reader.setResource(new ClassPathResource("static/person.csv"));
        // entity与csv数据做映射
        reader.setLineMapper(new DefaultLineMapper<Person>() {
            {
                setLineTokenizer(new DelimitedLineTokenizer() {
                    {
                        setNames("id", "name");
                    }
                });
                setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {
                    {
                        setTargetType(Person.class);
                    }
                });
            }
        });
        return reader;
    }

    /**
     * ItemWriter定义：指定datasource，设置批量插入sql语句，写入数据库
     *
     * @param dataSource
     * @return
     */
    @Bean
    public ItemWriter<Person> writer(DataSource dataSource) {
        // 使用jdbcBcatchItemWrite写数据到数据库中
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        // 设置有参数的sql语句
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
        String sql = "insert into person " + " (id,name) "
                + " values(:id,:name)";
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }


    @Bean
    public BatchJobListener batchJobListener() {
        return new BatchJobListener();
    }


    @Bean
    public ItemProcessor<Person, Person> validateProcessor() {
        BatchItemValidateProcessor validateProcessor = new BatchItemValidateProcessor();
        validateProcessor.setValidator(dataValidate());
        return validateProcessor;
    }

    @Bean
    public Validator<Person> dataValidate() {
        return new DataValidate<Person>();
    }

}
